package com.nxy.cpfsfe.util;


import com.adtec.starring.util.SpringUtil;
import com.nxy.cpfsfe.config.CpfsFeAllConfig;
import com.wangyin.r2m.client.jedis.JedisPubSub;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.Date;
import java.util.Map;



/**
 * 
 * jredis订阅者工具
 * @author nxygyzd
 *
 */
public class JredisUtil {
	//初始化对象
	private static JredisUtil jredisUtil = new JredisUtil();
	//构造函数私有化
	private JredisUtil (){
		
	}
	//公开获取单例对象函数
	public static JredisUtil  getInstance(){
		
		 return jredisUtil;
	}
	
	private final static Logger LOGGER = LoggerFactory.getLogger(JredisUtil.class);	
		
	
	/**
	 * 订阅者方法
	 * @param bizmap
	 * @return
	 */
	public  TimeoutableJedisPubSub redisSubscribe(Map<String, String> bizmap,long timtOut) {
		
		LOGGER.info("进入redis订阅者方法......");
		TimeoutableJedisPubSub timeoutableJedisPubSub = new TimeoutableJedisPubSub(timtOut,true,bizmap);
		LOGGER.info("进入redis消息接收,订阅通道为[{}]",bizmap.get("sndOrgNo")+bizmap.get("seqNb"));
		CpfsFeAllConfig.jedisCluster.subscribe(timeoutableJedisPubSub,  bizmap.get("sndOrgNo")+bizmap.get("seqNb"));
		LOGGER.info("redis订阅者工作完毕");
		return timeoutableJedisPubSub;
	
	}
	
	public class TimeoutableJedisPubSub extends JedisPubSub {
		
		private ThreadPoolTaskExecutor threadPool = (ThreadPoolTaskExecutor) SpringUtil.getBean("threadPoolTaskExecutor");
		
		public boolean flag;
		public String resMsg="";
		public Map<String, String> bizmap;
		
		
		public TimeoutableJedisPubSub(){
			super();
		}
		
		public TimeoutableJedisPubSub(long timeOut,boolean flag,Map<String, String> bizmap){
			
			this.flag = flag;
			this.bizmap = bizmap;
			threadPool.submit(new TimeoutUnsubscribeTask(this,timeOut));
			
		}
		
		@Override
		public void onMessage(String channel,String message){
			LOGGER.debug("当前时间[{}]，接收到通道[{}]的消息[{}]",DateUtil.getTimestamp(new Date()),channel,message);
			resMsg = message;
			this.unsubscribe(channel);
			LOGGER.debug("接收者已订阅到信息，取消此次通道[{}]的订阅",channel);
		}
		
	}
	
	public class TimeoutUnsubscribeTask implements Runnable{

		private TimeoutableJedisPubSub timeoutableJedisPubSub;
		
		private long timeOut;
		
		TimeoutUnsubscribeTask(){}
		
		TimeoutUnsubscribeTask(TimeoutableJedisPubSub timeoutableJedisPubSub,long timeOut){
			this.timeoutableJedisPubSub = timeoutableJedisPubSub;
			this.timeOut = timeOut;
		}
		
		@Override
		public void run() {
			try {
				Thread.currentThread().sleep(timeOut);
				timeoutableJedisPubSub.unsubscribe();
				LOGGER.debug("接收者订阅等待时间超时，取消此次通道[{}]的订阅",timeoutableJedisPubSub.bizmap.get("sndOrgNo")+timeoutableJedisPubSub.bizmap.get("seqNb"));
				timeoutableJedisPubSub.flag = false;
			} catch (InterruptedException e) {
				timeoutableJedisPubSub.unsubscribe();
				LOGGER.warn("接收者订阅服务异常，取消此次通道[{}]的订阅",timeoutableJedisPubSub.bizmap.get("sndOrgNo")+timeoutableJedisPubSub.bizmap.get("seqNb"));
				timeoutableJedisPubSub.flag = false;
			}
		}
		
	}
	
}
